/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.common.io;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;

import org.sss.util.BytesComparator;

/**
 * .
 * Module private
 */
class Worker {
  // if next == null, there is no more tuple.
  public byte[] next;
  public byte[] sentinel;
  public List<byte[]> bytes;
  public int index;
  private final KeyFormatter keyFormatter;
  public Worker(KeyFormatter keyFormatter) {
    this.next = null;
    this.sentinel = null;
    this.bytes = null;
    this.index = 0;
    this.keyFormatter = keyFormatter;
  }
  public boolean isEmpty() {
    return bytes == null || index >= bytes.size();
  }
  public byte[] firstKey() {
    return keyFormatter.getUserKey(bytes.get(index));
  }
} // class Worker

/**
 * .
 * Module private
 */
class WorkerQueue {
  private static BytesComparator CMP = new BytesComparator();
  private TreeMap<byte[], List<Worker>> map;
  public WorkerQueue() {
    this.map = new TreeMap<byte[], List<Worker>>(CMP);
  }
  public void push(byte[] key, Worker value) {
    List<Worker> workerList = map.get(key);
    if (workerList != null) {
      workerList.add(value);
    }
    else {
      workerList = new ArrayList<Worker>();
      workerList.add(value);
      map.put(key, workerList);
    }
  }
  public boolean isEmpty() {
    return map.isEmpty();
  }
  public Map.Entry<byte[], List<Worker>> pop() {
    Map.Entry<byte[], List<Worker>> e = map.firstEntry();
    map.remove(e.getKey());
    return e;
  }
  public void remove(byte[] key) {
    map.remove(key);
  }
} // class WorkerQueue
